package com.llx.dataset

import java.io.{BufferedReader, FileReader}

import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.scala.{ExecutionEnvironment, _}
import org.apache.flink.configuration.Configuration

import scala.collection.mutable.HashMap

/**
  *
  */
object Distribute_cache分布式缓存 {
  def main(args: Array[String]) {

    val env = ExecutionEnvironment.getExecutionEnvironment

    /**  一、设置分布式缓存文件位置
      * env.registerCachedFile
      * 本地IDE运行，缓存的是本地文件，缓存文件名字为cahce
      */
    env.registerCachedFile("file:///C://logs//flink", "cache")
    val data = env.fromElements("111", "222", "333")

    /**
      * 与广播类似
      * 通过继承map操作的RichMapFunction，或着先关操作的抽象类并实现
      * 在open中 获取缓存变量
      *
      */
    val result = data.map(new RichMapFunction[String, String] {
      val map = HashMap[String, String]()

      /** 二、获取缓存文件
        * getRuntimeContext.getDistributedCache.getFile("cache")
        * @param parameters
        */
      override def open(parameters: Configuration): Unit = {
        //getDistributedCache,getFile 来访问广播变量
        val file = getRuntimeContext.getDistributedCache.getFile("cache")//获取缓存文件
        //读取缓存文件内容到HashMap中，这个也可以使用广播实现
        val br = new BufferedReader(new FileReader(file))
        var line = br.readLine()

        while (line != null) {
          map.put(line, line + "" + line)
          line = br.readLine()
        }

      }

      override def map(value: String): String = {

        map.getOrElse(value, "default") //返回该value在map中的value值，如果不存在key为value的返回默认default

      }
    })

    result.print() //执行作业

  }

}
